package worker

import (
	"github.com/mongodb/mongo-go-driver/mongo"
	"../common"
	"github.com/mongodb/mongo-go-driver/mongo/clientopt"
	"context"
	"time"
)

//Mongodb存储日志
type LogSink struct {
	client         *mongo.Client       //mongodb客户端
	logCollection  *mongo.Collection   //数据表
	logChan        chan *common.JobLog //运输日志的channel
	autoCommitChan chan *common.LogBatch
}

//单例
var (
	G_logSink *LogSink
)

func InitLogSink() (err error) {
	var (
		client *mongo.Client
	)

	// 1, 建立连接
	if client, err = mongo.Connect(context.TODO(), G_config.MongodbUri,
		clientopt.ConnectTimeout(time.Duration(G_config.MongodbConnectTimeout)*time.Millisecond)); err != nil {
		return
	}

	//选择db和collection
	G_logSink = &LogSink{
		client:         client,
		logCollection:  client.Database("cron").Collection("log"),
		logChan:        make(chan *common.JobLog, 1000),
		autoCommitChan: make(chan *common.LogBatch, 1000),
	}

	//启动mongodb处理协程
	go G_logSink.writeLoop()
	return
}

//批量写入日志
func (logSink *LogSink) saveLogs(batch *common.LogBatch) {
	logSink.logCollection.InsertMany(context.TODO(), batch.Logs)
}

//日志存储协程
func (logSink *LogSink) writeLoop() {
	var (
		log          *common.JobLog
		logBatch     *common.LogBatch //当前批次
		commitTimer  *time.Timer
		timeoutBatch *common.LogBatch //超时批次
	)

	for {
		select {
		case log = <-logSink.logChan:
			//把这条log写到mongodb中,因为一条一条插入,需要等待一次mongodb的往返,
			//我们优化一下,批量插入,攒到多条之后在插入
			if logBatch == nil {
				logBatch = &common.LogBatch{}
				//让这个批次超时自动提交,给1s时间
				commitTimer = time.AfterFunc(time.Duration(G_config.JobLogCommitTimeout)*time.Millisecond,
					func(batch *common.LogBatch) func() {
						//必包,这样里面这个batch就和外面的batch无关了
						return func() {
							logSink.autoCommitChan <- batch
						}
					}(logBatch),
				)
			}
			//logSink/logCollection.insertOne()

			//把新的日志追加到批次中去
			logBatch.Logs = append(logBatch.Logs, log)

			//如果批次满了就立即发送
			if len(logBatch.Logs) >= G_config.JobLogBatchSize {
				//发送日志
				logSink.saveLogs(logBatch)
				//清空logbatch
				logBatch = nil
				//取消定时器
				commitTimer.Stop()
			}

		case timeoutBatch = <-logSink.autoCommitChan: //过期的批次
			//判断过期批次是否是当前批次
			if timeoutBatch != logBatch {
				continue //跳过已经被提交的批次
			}
			//把批次写入到mongodb中去
			logSink.saveLogs(timeoutBatch)
			//清空logbatch
			logBatch = nil
		}
	}
}

//发送日志
func (logSink *LogSink) Append(jobLog *common.JobLog) {
	select {
	case logSink.logChan <- jobLog:
	default:
		//队列满了就丢弃

	}

}
